package com.lagou.hbase.processor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTableWrapper;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;

import java.io.IOException;
import java.util.List;

public class DeleteProcessor extends BaseRegionObserver {
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit,
                           Durability durability) throws IOException {
        System.err.println("\n\n开始执行互关好友删除\n\n");
        // 监听表
        HTableWrapper relation = (HTableWrapper) e.getEnvironment().getTable(TableName.valueOf("relation"));
        // 获取删除对象
        List<Cell> cells = delete.getFamilyCellMap().get(Bytes.toBytes("friends"));
        // 判断是否为空
        if (CollectionUtils.isEmpty(cells)) {
            relation.close();
            return;
        }
        // 循环删除
        for (Cell cell : cells) {
            // 创建被删除好友，设置需要删除的 column
            Delete otherUserDelete = new Delete(CellUtil.cloneQualifier(cell));
            otherUserDelete.addColumns(Bytes.toBytes("friends"), CellUtil.cloneRow(cell));
            relation.delete(otherUserDelete);
            System.err.println(new String(CellUtil.cloneQualifier(cell)) + "删除" + new String(CellUtil.cloneRow(cell)) + "成功！！");
        }
        // 关闭 table 对象
        relation.close();
    }
}
